package com.whyx.project.openapi.Impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.reflect.TypeToken;

import com.whyx.project.openapi.config.GtApiConfig;
import exsun.jt808.data.base.StatusResult;
import exsun.jt808.data.config.CommonMsgTopic;
import exsun.jt808.data.config.jt808.Jt808ReceiveMsgTopic;
import exsun.jt808.data.distribute_data.DistributeEntity;
import exsun.jt808.data.distribute_data.jt1078.FileUploadReq;
import exsun.jt808.data.distribute_data.jt808.TakePhotoRightAway;
import exsun.jt808.data.distribute_data.jt808.Text;
import exsun.jt808.data.enume.OrderSendStatus;
import exsun.jt808.data.kafka_data.KafkaJT808Data;
import exsun.jt808.data.upload_data.GpsData;
import exsun.jt808.data.upload_data.Qingdao.DeviceVersionReporting;
import exsun.jt808.data.upload_data.Qingdao.UpgradeCompleteResponse;
import exsun.jt808.data.upload_data.jt1078.FileUploadNotice;
import exsun.jt808.data.upload_data.jt1078.MediaFileList;
import exsun.jt808.data.upload_data.jt808.*;
import exsun.jt808.data.util.*;
import exsun.jt808.data.util.parse.TimeUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.lang.reflect.Array;
import java.lang.reflect.Type;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

@Component
public class DistributeRepImpl {

    @Autowired
    GtApiConfig gtApiConfig;

    private final static Logger logger = LoggerFactory.getLogger(DistributeRepImpl.class);
    //key: requestId
    private static ConcurrentHashMap<String, DistributeEntity> requestEntityMap = new ConcurrentHashMap();
    private static SyncFutureUtil<DistributeRepImpl> syncFutureUtil
            = new SyncFutureUtil<DistributeRepImpl>(DistributeRepImpl.class, 500, 20 * 1000);

    @Value("${data.msgIds}")
    private String msgIds;


    public DistributeRepImpl() {
        Runnable runnable = new Runnable() {
            public void run() {
                try {
                    for (String key : requestEntityMap.keySet()) {
                        DistributeEntity items = requestEntityMap.get(key);
                        Long dt = items.getDistributeTime().getTime();
                        long now = new Date().getTime();
                        if (now - dt > 120000) {
                            removeRequestData(key);
                            logger.info("移除指令缓存数据:{},时间{} 剩余缓存{}条", key, TimeUtil.formatData(TimeUtil.dateFormatYMDHMS, dt), requestEntityMap.size());
                        }
                    }
                    logger.info("移除指令缓存数据:剩余缓存{}条", requestEntityMap.size());
                } catch (Exception e) {
                    logger.error("移除指令缓存数据异常{}", e.getMessage());
                }
            }
        };
        ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
        service.scheduleAtFixedRate(runnable, 10, 10, TimeUnit.SECONDS);
    }

    /**
     * 获取指令下发的应答
     *
     * @param key 接口调用处得到请求id
     * @return 应答结果
     * @throws Exception
     */
    public StatusResult getDistributeResponse(String key) throws Exception {
        StatusResult rep = InitSyncFuture(key, 8000);
        return rep;
    }

    public void addRequestData(String key, DistributeEntity entity) {
        requestEntityMap.put(key, entity);
    }

    private void removeRequestData(String key) {
        requestEntityMap.remove(key);
    }

    private String getKey(String requestId) {
        for (String key : requestEntityMap.keySet()) {
            if (requestEntityMap.get(key).requestId == null) continue;
            if (requestEntityMap.get(key).requestId.equals(requestId))
                return key;
        }
        return "";

    }

    /**
     * 网关执行下发指令的结果处理
     *
     * @param records
     * @param ack
     */
    @KafkaListener(topics = CommonMsgTopic.sendResultRData, containerFactory = "sendResultRDataFactory")
    public void sendResultRDataListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        logger.info("网关执行下发指令的结果应答：{}条", records.size());
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                logger.info("网关执行下发指令的结果应答:{}", msg);
                StatusResult r = GsonUtils.getObjectFromJson(msg, StatusResult.class);
                if (r.getResult() == null) continue;
                ObjectMapper objectMapper = new ObjectMapper();
                DistributeEntity distributeEntity = objectMapper.convertValue(r.getResult(), DistributeEntity.class);
                if (distributeEntity == null) continue;

                String id = getKey(distributeEntity.requestId);
                addRequestData(id, distributeEntity);

                String requestId = distributeEntity.requestId;
                boolean response = distributeEntity.response;
                if (requestId == null || (response == true && r.getCode() == OrderSendStatus.发送成功.ordinal())) {
                    continue;
                }
                String key = getKey(requestId);
                logger.info("########### 2 ########## {}", key);
                syncFutureUtil.setSyncFuture(key, r);
            } catch (Exception e) {
                logger.error("ResultRData处理异常：{}", e.getMessage());
            }

        }
        ack.acknowledge();
        logger.info("网关执行下发指令的结果应答处理完成");
    }

    @KafkaListener(topics = Jt808ReceiveMsgTopic.TerminalCommonResponse, containerFactory = "terminalCommonResponseFactory")
    public void terminalCommonResponseConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack){
        logger.info("收终端通用应答：{}条", records.size());
        com.whyx.project.openapi.map.MapUtil mapUtil= new com.whyx.project.openapi.map.MapUtil();
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                Type type = new TypeToken<KafkaJT808Data<TerminalCommonResponse>>(){}.getType();
                KafkaJT808Data<TerminalCommonResponse> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
                TerminalCommonResponse obj = kafkaJT808ReviveData.getObject();
                String[] split = msgIds.split(",");
                Integer[] integerStream = Arrays.stream(split).map(x -> Integer.parseInt(x)).toArray(Integer[]::new);
                boolean flag = false;
                for (int i = 0; i < integerStream.length ; i++) {
                    if (Integer.parseInt(String.valueOf(integerStream[i]),16) == obj.getOrgMsgId()){
                        flag = true;
                        break;
                    }
                }
                if (flag)continue;
                String key = getTerminalCommonResponse(obj);
                if(key == null)continue;
                StatusResult result = new StatusResult(OrderSendStatus.设备通用应答成功, obj);
                Map<String, Object> map = mapUtil.UpdateCommonDatabases(result);
                map.put("id",key);
                String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                String param1 = JSONObject.toJSONString(map);
                HttpUtil.send(url,param1, "UTF-8");
                syncFutureUtil.setSyncFuture(key, result);
            }catch (Exception e){
                logger.error("收终端通用应答处理异常：{}",e.getMessage());
            }
        }
        ack.acknowledge();
        logger.info("收终端通用应答处理完成");
    }


    @KafkaListener(topics = Jt808ReceiveMsgTopic.TakePhotoRightAwayResponse, containerFactory = "takePhotoRightAwayResponseFactory")
    public void takePhotoRightAwayResponseConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        logger.info("收到立即拍照应答：{}条", records.size());
        com.whyx.project.openapi.map.MapUtil mapUtil = new com.whyx.project.openapi.map.MapUtil();
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                Type type = new TypeToken<KafkaJT808Data<TakePhotoRightAwayResponse>>() {
                }.getType();
                KafkaJT808Data<TakePhotoRightAwayResponse> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
                TakePhotoRightAwayResponse obj = kafkaJT808ReviveData.getObject();
                String key = getTakePhotoRightAwayResponse(obj);
                if (key == null) continue;

                StatusResult result = new StatusResult(OrderSendStatus.设备特殊应答成功, obj);
                Map<String, Object> map = mapUtil.UpdateDatabases(result);
                map.put("id", key);
                String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                String param1 = JSONObject.toJSONString(map);
                HttpUtil.send(url, param1, "UTF-8");
                syncFutureUtil.setSyncFuture(key, result);
            } catch (Exception e) {
                logger.error("收到立即拍照应答处理异常：{}", e.getMessage());
            }

        }
        ack.acknowledge();
        logger.info("收到立即拍照应答处理完成");
    }

    /**
     * 查询终端参数应答处理
     *
     * @param records
     * @param ack
     */
    @KafkaListener(topics = Jt808ReceiveMsgTopic.TerminalQueryParamResponse, containerFactory = "TerminalQueryParamResponseFactory")
    public void terminalQueryParamResponseConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        logger.info("收到查询终端参数应答：{}条", records.size());
        com.whyx.project.openapi.map.MapUtil mapUtil = new com.whyx.project.openapi.map.MapUtil();
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                Type type = new TypeToken<KafkaJT808Data<TerminalParameters>>() {
                }.getType();
                KafkaJT808Data<TerminalParameters> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
                TerminalParameters obj = kafkaJT808ReviveData.getObject();
                String key = getTerminalParameters(obj);
                if (key == null) continue;
                StatusResult result = new StatusResult(OrderSendStatus.设备特殊应答成功, obj);
                Map<String, Object> map = mapUtil.UpdateDatabases(result);
                map.put("id", key);
                map.put("responseResult", obj);
                String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                String param1 = JSONObject.toJSONString(map);
                HttpUtil.send(url, param1, "UTF-8");
                syncFutureUtil.setSyncFuture(key, result);
            } catch (Exception e) {
                logger.error("收到查询终端参数应答处理异常：{}", e.getMessage());
            }
        }
        ack.acknowledge();
        logger.info("收到查询参数应答处理完成");
    }

    /**
     * 查询指定终端参数应答处理
     *
     * @param records
     * @param ack
     */
    @KafkaListener(topics = Jt808ReceiveMsgTopic.QuerySpecifiedParametersResponse, containerFactory = "QuerySpecifiedParametersResponseFactory")
    public void terminalQuerySpecifiedParamResponseConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        logger.info("收到查询指定终端参数应答：{}条", records.size());
        com.whyx.project.openapi.map.MapUtil mapUtil = new com.whyx.project.openapi.map.MapUtil();
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                Type type = new TypeToken<KafkaJT808Data<TerminalParameters>>() {
                }.getType();
                KafkaJT808Data<TerminalParameters> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
                TerminalParameters obj = kafkaJT808ReviveData.getObject();
                String key = getQuerySpecifiedParameters(obj);
                if (key == null) continue;
                StatusResult result = new StatusResult(OrderSendStatus.设备特殊应答成功, obj);
                Map<String, Object> map = mapUtil.UpdateDatabases(result);
                map.put("id", key);
                String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                String param1 = JSONObject.toJSONString(map);
                HttpUtil.send(url, param1, "UTF-8");
                syncFutureUtil.setSyncFuture(key, result);
            } catch (Exception e) {
                logger.error("收到查询指定终端参数应答处理异常：{}", e.getMessage());
            }
        }
        ack.acknowledge();
        logger.info("收到查询指定终端参数应答处理完成");
    }

    @KafkaListener(topics = Jt808ReceiveMsgTopic.DeviceVersionReporting, containerFactory = "deviceVersionReportingResponseFactory")
    public void DeviceVersionReportingResponseConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        logger.info("收到设备版本查询应答：{}条", records.size());
        com.whyx.project.openapi.map.MapUtil mapUtil = new com.whyx.project.openapi.map.MapUtil();
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                Type type = new TypeToken<KafkaJT808Data<DeviceVersionReporting>>() {
                }.getType();
                KafkaJT808Data<DeviceVersionReporting> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
                DeviceVersionReporting obj = kafkaJT808ReviveData.getObject();
                String key = getDeviceVersionReporting(obj);
                if (key == null) continue;
                StatusResult result = new StatusResult(OrderSendStatus.设备特殊应答成功, obj);
                Map<String, Object> map = mapUtil.UpdateDatabases(result);
                map.put("id", key);
                String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                String param1 = JSONObject.toJSONString(map);
                HttpUtil.send(url, param1, "UTF-8");
                syncFutureUtil.setSyncFuture(key, result);
            } catch (Exception e) {
                logger.error("收到设备版本查询应答处理异常：{}", e.getMessage());
            }
        }
        ack.acknowledge();
        logger.info("收到设备版本查询应答处理完成");
    }


    @KafkaListener(topics = Jt808ReceiveMsgTopic.QueryPositionInfoResponse, containerFactory = "QueryPositionInfoResponseFactory")
    public void QueryPositionInfoResponseConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        logger.info("收到位置信息查询应答：{}条", records.size());
        com.whyx.project.openapi.map.MapUtil mapUtil = new com.whyx.project.openapi.map.MapUtil();
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                Type type = new TypeToken<KafkaJT808Data<GpsData>>() {
                }.getType();
                KafkaJT808Data<GpsData> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
                GpsData obj = kafkaJT808ReviveData.getObject();
                String key = getPositionProperty(obj);
                if (key == null) continue;
                StatusResult result = new StatusResult(OrderSendStatus.设备特殊应答成功, obj);
                Map<String, Object> map = mapUtil.UpdateDatabases(result);
                map.put("id", key);
                String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                String param1 = JSONObject.toJSONString(map);
                HttpUtil.send(url, param1, "UTF-8");
                syncFutureUtil.setSyncFuture(key, result);
            } catch (Exception e) {
                logger.error("收到位置信息查询处理异常：{}", e.toString());
            }
        }
        ack.acknowledge();
        logger.info("收到位置信息查询处理完成");
    }

    @KafkaListener(topics = Jt808ReceiveMsgTopic.TerminalQueryPropResponse, containerFactory = "TerminalQueryPropResponseFactory")
    public void TerminalQueryPropResponseConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        logger.info("收到查询终端属性应答：{}条", records.size());
        com.whyx.project.openapi.map.MapUtil mapUtil = new com.whyx.project.openapi.map.MapUtil();
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                Type type = new TypeToken<KafkaJT808Data<TerminalProperty>>() {
                }.getType();
                KafkaJT808Data<TerminalProperty> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
                TerminalProperty obj = kafkaJT808ReviveData.getObject();
                String key = getTerminalProperty(obj);
                if (key == null) continue;
                StatusResult result = new StatusResult(OrderSendStatus.设备特殊应答成功, obj);
                Map<String, Object> map = mapUtil.UpdateDatabases(result);
                map.put("id", key);
                String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                String param1 = JSONObject.toJSONString(map);
                HttpUtil.send(url, param1, "UTF-8");
                syncFutureUtil.setSyncFuture(key, result);
            } catch (Exception e) {
                logger.error("收到查询终端属性应答处理异常：{}", e.getMessage());
            }
        }
        ack.acknowledge();
        logger.info("收到查询终端属性应答处理完成");
    }

    @KafkaListener(topics = Jt808ReceiveMsgTopic.FileUploadResponse, containerFactory = "FileUploadResponseFactory")
    public void FileUploadResponseConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        logger.info("收到文件上传完成应答：{}条", records.size());
        com.whyx.project.openapi.map.MapUtil mapUtil = new com.whyx.project.openapi.map.MapUtil();
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                Type type = new TypeToken<KafkaJT808Data<FileUploadNotice>>() {
                }.getType();
                KafkaJT808Data<FileUploadNotice> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
                FileUploadNotice obj = kafkaJT808ReviveData.getObject();
                String key = getFileUploadResponse(obj);
                if (key == null) continue;
                if (obj.getUploadResult() == 0){
                    logger.info("进入设备应答成功的判断");
                    StatusResult result = new StatusResult(OrderSendStatus.设备特殊应答成功, obj);
                    Map<String, Object> map = mapUtil.UpdateDatabases(result);
                    map.put("id", key);
                    String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                    String param1 = JSONObject.toJSONString(map);
                    HttpUtil.send(url, param1, "UTF-8");
                    syncFutureUtil.setSyncFuture(key, result);
                }else {
                    logger.info("进入设备应答失败的判断");
                    StatusResult result = new StatusResult(OrderSendStatus.设备应答失败, obj);
                    Map<String, Object> map = mapUtil.UpdateDatabases(result);
                    map.put("id", key);
                    String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                    String param1 = JSONObject.toJSONString(map);
                    HttpUtil.send(url, param1, "UTF-8");
                    syncFutureUtil.setSyncFuture(key, result);
                }

            } catch (Exception e) {
                logger.error("收到文件上传完成应答处理异常：{}", e.getMessage());
            }
        }
        ack.acknowledge();
        logger.info("收到文件上传完成应答处理完成");
    }


    @KafkaListener(topics = Jt808ReceiveMsgTopic.UpgradeResponse, containerFactory = "UpgradeCompleteResponseFactory")
    public void UpgradeCompleteResponse_SendConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        logger.info("收到远程升级应答：{}条", records.size());
        com.whyx.project.openapi.map.MapUtil mapUtil = new com.whyx.project.openapi.map.MapUtil();
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                Type type = new TypeToken<KafkaJT808Data<UpgradeCompleteResponse>>() {
                }.getType();
                KafkaJT808Data<UpgradeCompleteResponse> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
                UpgradeCompleteResponse obj = kafkaJT808ReviveData.getObject();
                String key = getUpgradeCompleteResponse(obj);
                logger.info(key);
                if (key == null) continue;
                StatusResult result = new StatusResult(OrderSendStatus.设备特殊应答成功, obj);
//                JSONObject json = JSONObject.parseObject(msg);
//                String devNo = String.valueOf(json.get("devNo"));
//                JSONObject jsonObject = JSONObject.parseObject(String.valueOf(json.get("object")));
//                String vNumber = String.valueOf(jsonObject.get("VNumber"));
//                Map<String, Object> map = mapUtil.UpdateDatabases(result, vNumber, devNo);
                Map<String, Object> map = mapUtil.UpdateDatabases(result);
                map.put("id", key);
                String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                String param1 = JSONObject.toJSONString(map);
                logger.info(param1);
                HttpUtil.send(url, param1, "UTF-8");
                syncFutureUtil.setSyncFuture(key, result);
            } catch (Exception e) {
                logger.error("收到远程升级应答处理异常：{}", e.getMessage());
            }
        }
        ack.acknowledge();
        logger.info("收到远程升级应答处理完成");

    }

    @KafkaListener(topics = Jt808ReceiveMsgTopic.MediaFileList, containerFactory = "MediaFileListFactory")
    public void MediaFileListConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack)  {
        logger.info("收到终端上传音视频资源列表应答：{}条", records.size());
        com.whyx.project.openapi.map.MapUtil mapUtil= new com.whyx.project.openapi.map.MapUtil();
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                Type type = new TypeToken<KafkaJT808Data<FileUploadNotice>>(){}.getType();
                KafkaJT808Data<MediaFileList> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
                MediaFileList obj = kafkaJT808ReviveData.getObject();
                String key = getMediaFileList(obj);
                if(key == null)continue;
                StatusResult result = new StatusResult(OrderSendStatus.设备特殊应答成功, obj);
                Map<String, Object> map = mapUtil.UpdateDatabases(result);
                map.put("id",key);
                String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                String param1 = JSONObject.toJSONString(map);
                HttpUtil.send(url,param1, "UTF-8");
                syncFutureUtil.setSyncFuture(key, result);
            }catch (Exception e){
                logger.error("收到终端上传音视频资源列表应答处理异常：{}",e.getMessage());
            }
        }
        ack.acknowledge();
        logger.info("收到终端上传音视频资源列表应答处理完成");
    }

    @KafkaListener(topics = Jt808ReceiveMsgTopic.AutoCalibrationResponse, containerFactory = "AutoCalibrationResponseFactory")
    public void AutoCalibrationResponseConsumerListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        logger.info("收到农机终端自动标定应答：{}条", records.size());
        com.whyx.project.openapi.map.MapUtil mapUtil = new com.whyx.project.openapi.map.MapUtil();
        for (ConsumerRecord<?, ?> record : records) {
            try {
                String msg = record.value().toString();
                Type type = new TypeToken<KafkaJT808Data<AutoCalibrationResponse>>() {
                }.getType();
                KafkaJT808Data<AutoCalibrationResponse> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
                AutoCalibrationResponse obj = kafkaJT808ReviveData.getObject();
                String key = getAutoCalibrationResponse(obj);
                if (key == null) continue;
                if (obj.getFlag() == 0){
                    logger.info("进入设备应答成功的判断");
                    StatusResult result = new StatusResult(OrderSendStatus.设备特殊应答成功, obj);
                    Map<String, Object> map = mapUtil.UpdateDatabases(result);
                    map.put("id", key);
                    String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                    String param1 = JSONObject.toJSONString(map);
                    HttpUtil.send(url, param1, "UTF-8");
                    syncFutureUtil.setSyncFuture(key, result);
                }else {
                    logger.info("进入设备应答失败的判断");
                    StatusResult result = new StatusResult(OrderSendStatus.设备应答失败, obj);
                    Map<String, Object> map = mapUtil.UpdateDatabases(result);
                    map.put("id", key);
                    String url = gtApiConfig.getOrderSendRecordUpdateUrl();
                    String param1 = JSONObject.toJSONString(map);
                    HttpUtil.send(url, param1, "UTF-8");
                    syncFutureUtil.setSyncFuture(key, result);
                }
            } catch (Exception e) {
                logger.error("收到农机终端自动标定应答处理异常：{}", e.getMessage());
            }
        }
        ack.acknowledge();
        logger.info("收到农机终端自动标定应答处理完成");
    }



    private String getMediaFileList(MediaFileList rep){
        return matchByMsgIdAndSerialNo(rep.dvo,0x9205,rep.getOrderSerialNo());
    }


    private String getTerminalCommonResponse(TerminalCommonResponse rep) {
        return matchByMsgIdAndSerialNo(rep.dvo, rep.getOrgMsgId(), rep.getOrgSerialNum());
    }

    private String getTakePhotoRightAwayResponse(TakePhotoRightAwayResponse rep) {
        return matchByMsgIdAndSerialNo(rep.dvo, 0x8801, rep.getRepSerialNo());
    }

    private String getQuerySpecifiedParameters(TerminalParameters rep) {
        return matchByMsgIdAndSerialNo(rep.dvo, 0x8106, rep.getSerialNo());
    }

    private String getTerminalParameters(TerminalParameters rep) {
        return matchByMsgIdAndSerialNo(rep.dvo, 0x8104, rep.getSerialNo());
    }

    private String getDeviceVersionReporting(DeviceVersionReporting rep) {
        return matchByMsgId(rep.dvo, 0x8902);
    }

    private String getTerminalProperty(TerminalProperty rep) {
        return matchByMsgIdAndSerialNo(rep.dvo, 0x8107, rep.getMsgSerialNo());
    }

    private String getPositionProperty(GpsData rep) {
        return matchByMsgIdAndSerialNo(rep.dvo, 0x8201, rep.getSerialNo());
    }

    private String getUpgradeCompleteResponse(UpgradeCompleteResponse rep) {
        return matchByMsgId(rep.dvo, 0x8901);
    }

    private String getFileUploadResponse(FileUploadNotice rep) {
        return matchByMsgIdAndSerialNo(rep.dvo, 0x9206, rep.getResponSerialNum());
    }
    private String getAutoCalibrationResponse(AutoCalibrationResponse rep) {
        return matchByMsgId(rep.dvo, 0x8903);
    }

    private String matchByMsgIdAndSerialNo(String devNo, int repMsgId, int repSerialNum) {
        String result = null;
        for (String requestId : requestEntityMap.keySet()) {
            DistributeEntity distributeEntity = requestEntityMap.get(requestId);
            if (distributeEntity.devNo.equals(devNo)
                    && distributeEntity.getMsgId() == repMsgId
                    && distributeEntity.getSerialNo() == repSerialNum) {
                result = requestId;
                break;
            }
        }

        return result;
    }

    private String matchByMsgId(String devNo, int repMsgId) {
        String result = null;
        for (String requestId : requestEntityMap.keySet()) {
            DistributeEntity distributeEntity = requestEntityMap.get(requestId);
            if (distributeEntity.devNo.equals(devNo)
                    && distributeEntity.getMsgId() == repMsgId) {
                result = requestId;
                break;
            }
        }
        return result;
    }

    private StatusResult InitSyncFuture(String requestId, long timeout) throws Exception {
        syncFutureUtil.removeSyncFuture(requestId);
        SyncFuture<Object> syncFuture = syncFutureUtil.getSyncFuture(requestId);
        //此处会锁住阻塞，直到有应答或超时解锁
        logger.info("########## 1 ############ {}", requestId);
        StatusResult result = (StatusResult) syncFuture.get(timeout, TimeUnit.MILLISECONDS);
        logger.info("########## 3 ############ {}", requestId);
        logger.info("result:---" + result);
        syncFutureUtil.removeSyncFuture(requestId);
//        removeRequestData(requestId);
        return result;
    }

    public SyncFuture InitSyncFuture(String requestId)
    {
        syncFutureUtil.removeSyncFuture(requestId);
        SyncFuture<Object> syncFuture = syncFutureUtil.getSyncFuture(requestId);
        return  syncFuture;
    }


    public StatusResult getResult(String requestId, long timeout,SyncFuture syncFuture) throws Exception
    {
        //此处会锁住阻塞，直到有应答或超时解锁
        logger.info("########## 1 ############ {}",requestId);
        StatusResult result = (StatusResult)syncFuture.get(timeout, TimeUnit.MILLISECONDS);
        logger.info("########## 3 ############ {}",requestId);
        syncFutureUtil.removeSyncFuture(requestId);
//        removeRequestData(requestId);
        return result;
    }
}
